package com.uam.core.lk.utils.kafka;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * @author 祝展
 * @version 1.0
 * @description
 * @Date 2020/1/15 18:42
 **/
public class KafkaSerializer<T> implements Serializer<T> {

    private ObjectMapper objectMapper;

    public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;

    public KafkaSerializer(){
        ObjectMapper objectMapper = new ObjectMapper();
        // 序列化的范围，字段的可见级别，ANY标识任意。
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        // 序列化的时候，序列化结果里面包含类型信息。
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        // 过滤掉 null的属性
        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        // 关闭FAIL_ON_EMPTY_BEANS， 告诉Jackson空对象不要抛异常
        objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS);
        // 反序列化的时候，遇到未知属性不要报异常
        objectMapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
        // 如果enable，则：视空字符传为null，如果disable则空字符串就是空字符串
        objectMapper.disable(DeserializationFeature.ACCEPT_EMPTY_STRING_AS_NULL_OBJECT);
        this.objectMapper = objectMapper;
    }

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
    }

    @Override
    public byte[] serialize(String s, T t) {
            try {
                byte[] bytes =  t == null ? null : this.objectMapper.writeValueAsBytes(t);
                return bytes;
            } catch (JsonProcessingException e) {
                e.printStackTrace();
                SerializationException exception =  new SerializationException("Error when KafkaSerializer serializing Object to byte[] due to unsupported",e);
                throw  exception;
            }
    }

    @Override
    public void close() {
    }
}
